﻿using Acl.Messages;
using Acl.Messages.Impl;

using System.Buffers;
using System.Runtime.CompilerServices;

namespace Acl.Messages;

/// <summary>
/// 分布式总线
/// </summary>
public class DistributionMessageBus : IDistributionMessageBus
{
    private readonly Func<NatsConnection> _connectionFactory;
    private readonly IDistributionMessageHandleManagement _messageManagement;

    /// <summary>
    /// 
    /// </summary>
    public IDistributionMessageHandleManagement MessageHandleManagement => _messageManagement;

    /// <summary>
    /// 构造分布式总线
    /// </summary>
    /// <param name="connectionFactory">连接工厂</param>
    /// <param name="serviceProvider"></param>
    public DistributionMessageBus(Func<NatsConnection> connectionFactory, IServiceProvider? serviceProvider = default)
    {
        _connectionFactory = connectionFactory;
        _messageManagement = new DistributionMessageHandleManagement(this, serviceProvider);
    }



    /// <summary>
    ///
    /// </summary>
    /// <param name="topic"></param>
    /// <param name="payload"></param>
    /// <param name="opts"></param>
    /// <param name="cancellationToken"></param>
    /// <returns></returns>
    public ValueTask Publish(string topic,  in DistributionPublishOptions? opts = null, CancellationToken cancellationToken = default)
    {
        return new NatsPublisher(_connectionFactory()).Publish(topic,  opts, cancellationToken);
    }

    /// <summary>
    ///
    /// </summary>
    /// <typeparam name="T"></typeparam>
    /// <param name="topic"></param>
    /// <param name="data"></param>
    /// <param name="opts"></param>
    /// <param name="cancellationToken"></param>
    /// <returns></returns>
    public ValueTask Publish<T>(string topic, T? data = default, in DistributionPublishOptions? opts = null, CancellationToken cancellationToken = default)
    {
        return new NatsPublisher(_connectionFactory()).Publish(topic, data, opts, cancellationToken);
    }

    private readonly struct NatsPublisher(INatsConnection conn) : IDistributionMessagePublisher
    {
        private readonly INatsConnection _conn = conn;

        public ValueTask Publish(string topic,  in DistributionPublishOptions? opts = null, CancellationToken cancellationToken = default)
        {
            var (natsOps,headers,replyTo) =   opts.ToNatsPubOps();

            return _conn.PublishAsync(topic, headers,replyTo, natsOps, cancellationToken);
        }

        public ValueTask Publish<T>(string topic, T? data = default, in DistributionPublishOptions? opts = null, CancellationToken cancellationToken = default)
        {
            var (natsOps, headers, replyTo) = opts.ToNatsPubOps();

            if (data is byte[] bytes)
            {
                return _conn.PublishAsync(topic, bytes,headers, replyTo, serializer:default, natsOps, cancellationToken);
            }
            else if (data is ReadOnlyMemory<byte> memory)
            {
                return _conn.PublishAsync(topic, memory, headers, replyTo, serializer: default, natsOps, cancellationToken);
            }
            else if (data is IEnumerable<byte> stream)
            {
                Memory<byte> buff = stream.ToArray();
                return _conn.PublishAsync(topic, buff, headers, replyTo, serializer: default, natsOps, cancellationToken);
            }

            return _conn.PublishAsync<T>(topic, data!, headers, replyTo, serializer: default, natsOps, cancellationToken);
        }

    }


    /// <summary>
    ///
    /// </summary>
    /// <typeparam name="TData"></typeparam>
    /// <param name="topic"></param>
    /// <param name="action"></param>
    /// <param name="options"></param>
    /// <param name="cancellationToken"></param>
    /// <returns></returns>
    public async Task Subscribe<TData>(string topic, Action<IDistributionMessage<TData>> action, DistributionSubscribeOptions? options = null, CancellationToken cancellationToken = default)
    {
        var conn = _connectionFactory();
        var (natsOps, queueGroup) = options.ToNatsSubOpts();

        _ = Task.Run(async () =>
          {
              await foreach (var msg in conn.SubscribeAsync<TData>(topic, queueGroup, serializer: default, natsOps, cancellationToken))
              {
                  var message = new DistributedMessage<TData>
                  {
                      ReplyTo = msg.ReplyTo,
                      Topic = msg.Subject,
                      Headers = msg.Headers,
                      Data = msg.Data,
                      Size = msg.Size,
                      Publisher = new NatsPublisher(msg.Connection!),
                  };

                  action(message);
              }

              Console.WriteLine($"unsubscribe:{topic}");
          });

        await conn.PingAsync().ConfigureAwait(false);
    }

#pragma warning disable CS8613 // 返回类型中引用类型的为 Null 性与隐式实现的成员不匹配。

    /// <summary>
    ///
    /// </summary>
    /// <typeparam name="TRequest"></typeparam>
    /// <typeparam name="TReplay"></typeparam>
    /// <param name="topic"></param>
    /// <param name="data"></param>
    /// <param name="reqestOpts"></param>
    /// <param name="replyOpts"></param>
    /// <param name="cancellationToken"></param>
    /// <returns></returns>
    public async ValueTask<IDistributionMessage<TReplay>?> Send<TRequest, TReplay>(string topic, TRequest data, DistributionPublishOptions? reqestOpts, DistributionSubscribeOptions? replyOpts, CancellationToken cancellationToken)
#pragma warning restore CS8613 // 返回类型中引用类型的为 Null 性与隐式实现的成员不匹配。
    {
        var conn = _connectionFactory();
        var (natsReqOps, headers, _) = reqestOpts.ToNatsPubOps();
        var (natsReplyOpts, _) = replyOpts.ToNatsSubOpts();

        var natsMsg = await conn.RequestAsync<TRequest, TReplay>(topic, data,headers, requestSerializer:default, replySerializer:default, natsReqOps, natsReplyOpts,  cancellationToken).ConfigureAwait(false);


        return new DistributedMessage<TReplay>
        {
            Data = natsMsg.Data,
            Headers = natsMsg.Headers,
            ReplyTo = natsMsg.ReplyTo,
            Topic = natsMsg.Subject,
            Size = natsMsg.Size,
        };
    }

    /// <summary>
    ///
    /// </summary>
    /// <typeparam name="TRequest"></typeparam>
    /// <typeparam name="TReply"></typeparam>
    /// <param name="topic"></param>
    /// <param name="data"></param>
    /// <param name="reqestOpts"></param>
    /// <param name="replyOpts"></param>
    /// <param name="cancellationToken"></param>
    /// <returns></returns>
    public async IAsyncEnumerable<IDistributionMessage<TReply?>> RequestManyAsync<TRequest, TReply>(string topic, TRequest? data, DistributionPublishOptions? reqestOpts = null, DistributionSubscribeOptions? replyOpts = null, [EnumeratorCancellation] CancellationToken cancellationToken = default)
    {
        var conn = _connectionFactory();
        var (natsReqOps, headers, _) = reqestOpts.ToNatsPubOps();
        var (natsReplyOpts, _) = replyOpts.ToNatsSubOpts();

        var msgs = conn.RequestManyAsync<TRequest, TReply>(topic, data,headers, requestSerializer: default, replySerializer: default, natsReqOps, natsReplyOpts,cancellationToken);

        await foreach (var natsMsg in msgs)
        {
            yield return new DistributedMessage<TReply>
            {
                Data = natsMsg.Data,
                Headers = natsMsg.Headers,
                ReplyTo = natsMsg.ReplyTo,
                Topic = natsMsg.Subject,
                Size = natsMsg.Size,
            };
        }
    }
}